package com.cetc.sdp.kmga.cs.jdbc;

import com.cetc.sdp.kmga.cs.audit.*;
import com.cetc.sdp.kmga.cs.common.IndexTable;
import com.cetc.sdp.kmga.cs.common.TableMapping;
import com.cetc.sdp.kmga.cs.common.TableMeta;
import com.cetc.sdp.kmga.cs.util.Tool;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.DayOfWeek;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * @description:
 * @author： DengQiang
 * @date: 2017/8/29 10:24
 */
public class AuditDAO {

    /**
     * 加载hbase表，元数据信息
     *
     * @param topic
     * @return
     */
    public static Optional<TableMapping> findTableMetaInfoByTopic(String topic) {
        String sql = "SELECT a.table_id, table_name, rowkey_type, topic, queue, log_type, append_create_time, append_suffix_by_month, repartition, num_partition, " +
                " split, initial_value, current_value, update_time, column_name, column_family, mapping_index, is_row_key, inner_index" +
                " FROM sdp_kmga_cs_hbase_log_topic_meta_table AS a LEFT JOIN sdp_kmga_cs_hbase_meta_table_column_mapping AS b" +
                " ON a.table_id=b.table_id WHERE a.topic=?";
        return findTableMetaInfoByKey(sql, topic);
    }

    /**
     * 加载hbase表，元数据信息
     *
     * @param queue
     * @return
     */
    public static Optional<TableMapping> findTableMetaInfoByQueue(String queue) {
        String sql = "SELECT a.table_id, table_name, rowkey_type, topic, queue, log_type, append_create_time, append_suffix_by_month, repartition, num_partition, " +
                " split, initial_value, current_value, update_time, column_name, column_family, mapping_index, is_row_key, inner_index" +
                " FROM sdp_kmga_cs_hbase_log_topic_meta_table AS a LEFT JOIN sdp_kmga_cs_hbase_meta_table_column_mapping AS b" +
                " ON a.table_id=b.table_id WHERE a.queue=?";
        return findTableMetaInfoByKey(sql, queue);
    }

    public static TableMeta findTableMetaByName(String tableName) {
        String sql = "SELECT table_id, table_name, rowkey_type, topic, queue, log_type, " +
                " append_create_time, append_suffix_by_month, repartition, num_partition, " +
                " split, initial_value, current_value, update_time FROM " +
                " sdp_kmga_cs_hbase_log_topic_meta_table WHERE table_name=?";
        Optional<TableMeta> res = SQLHelper.find(sql, stmt -> {
            stmt.setString(1, tableName);
            return stmt.executeQuery();
        }, rs -> {
            TableMeta tm = new TableMeta();
            boolean flag = false;
            while (rs.next()) {
                flag = true;
                tm.setId(rs.getInt("table_id"));
                tm.setTableName(rs.getString("table_name"));
                tm.setRowkeyType(rs.getInt("rowkey_type"));
                tm.setTopic(rs.getString("topic"));
                tm.setQueue(rs.getString("queue"));
                tm.setLogType(rs.getString("log_type"));
                tm.setGenerateCreateTime(rs.getInt("append_create_time"));
                tm.setAutoAppendSuffix(rs.getInt("append_suffix_by_month"));
                tm.setRepartition(rs.getBoolean("repartition"));
                tm.setNumOfPartition(rs.getInt("num_partition"));
                tm.setInitial(rs.getLong("initial_value"));
                tm.setCurrent(rs.getLong("current_value"));
                tm.setSplit(rs.getBoolean("split"));
                tm.setLastUpdateTime(LocalDateTime.from(Tool.DEFAULT_TIME_FORMATTER.parse(rs.getString("update_time"))));
            }
            return flag ? tm : null;
        });

        return res.get();
    }

    private static Optional<TableMapping> findTableMetaInfoByKey(String sql, String val) {
        Optional<TableMapping> res = SQLHelper.find(sql, stmt -> {
            stmt.setString(1, val);
            return stmt.executeQuery();
        }, rs -> {
            TableMapping tm = new TableMapping();
            boolean flag = false;
            while (rs.next()) {
                flag = true;
                tm.setId(rs.getInt("table_id"));
                tm.setTableName(rs.getString("table_name"));
                tm.setRowkeyType(rs.getInt("rowkey_type"));
                tm.setTopic(rs.getString("topic"));
                tm.setQueue(rs.getString("queue"));
                tm.setLogType(rs.getString("log_type"));
                tm.setGenerateCreateTime(rs.getInt("append_create_time"));
                tm.setAutoAppendSuffix(rs.getInt("append_suffix_by_month"));
                tm.setRepartition(rs.getBoolean("repartition"));
                tm.setNumOfPartition(rs.getInt("num_partition"));
                tm.setInitial(rs.getLong("initial_value"));
                tm.setCurrent(rs.getLong("current_value"));
                tm.setSplit(rs.getBoolean("split"));
                tm.setLastUpdateTime(LocalDateTime.from(Tool.DEFAULT_TIME_FORMATTER.parse(rs.getString("update_time"))));
                TableMapping.Column column = new TableMapping.Column();
                column.setColumnName(rs.getString("column_name"));
                column.setColumnFamily(rs.getString("column_family"));
                column.setMappingIndex(rs.getInt("mapping_index"));
                column.setIsRowKey(rs.getInt("is_row_key"));
                column.setInnerIdex(rs.getInt("inner_index"));
                tm.addColumn(column);
                if (tm.isSplit()) {
                    tm.setSplitTables(findSplitTableByTableId(tm.getId()));
                }
            }
            return flag ? tm : null;
        });
        res.ifPresent(t -> {
            List<IndexTable> indexTables = findIndexTableByTableId(t.getId());
            indexTables.forEach(it -> {
                String colName = it.getIndexColumnName();
                it.setIndexColumn(-1);
                if (colName != null) {
                    TableMapping.Column column = t.getColumn(colName);
                    if (column != null) {
                        it.setIndexColumn(column.getMappingIndex());
                    }
                }
            });
            if (t.isSplit()) {
                Map<String, List<IndexTable>> listMap = new HashMap<>(16);
                indexTables.forEach(indexTable -> {
                    List<IndexTable> list = listMap.computeIfAbsent(indexTable.getSplitKey(), k -> new ArrayList<>());
                    list.add(indexTable);
                });
                t.setSplitIndexTables(listMap);
            } else {
                t.setIndexTables(indexTables);
            }
        });
        return res;
    }

    public static List<IndexTable> findIndexTableByTableId(int tableId) {
        String sql = "SELECT index_table_name, column_index, rowkey_len, subkey_len, random_len, append_char, split_key FROM sdp_kmga_cs_hbase_index_table WHERE table_id=?";
        List<IndexTable> res = new ArrayList<>();
        SQLHelper.find(sql, st -> {
            st.setInt(1, tableId);
            return st.executeQuery();
        }, rs -> {
            while (rs.next()){
                IndexTable table = new IndexTable();
                table.setTableName(rs.getString("index_table_name"));
                table.setIndexColumnName(rs.getString("column_index"));
                table.setRowkeyLen(rs.getInt("rowkey_len"));
                table.setSubkeyLen(rs.getInt("subkey_len"));
                table.setAppendChar(rs.getString("append_char"));
                table.setSplitKey(rs.getString("split_key"));
                table.setRandomLen(rs.getInt("random_len"));
                res.add(table);
            }
            return null;
        });
        return res;
    }

    public static Map<String, String> findSplitTableByTableId(int tableId) {
        String sql = "SELECT table_name, split_key FROM sdp_kmga_cs_hbase_split_table WHERE table_id=?";
        Map<String, String> splitTables = new HashMap<>(16);
        SQLHelper.find(sql, st -> {
            st.setInt(1, tableId);
            return st.executeQuery();
        }, rs -> {
            while (rs.next()) {
                splitTables.put(rs.getString("split_key"), rs.getString("table_name"));
            }
            return null;
        });
        return splitTables;
    }

    public static void updateHbaseRecordSeqValue(String tableName, long current) {
        String sql = "UPDATE sdp_kmga_cs_hbase_log_topic_meta_table SET current_value=?, update_time=? WHERE table_name = ?";
        SQLHelper.save(sql, ps -> {
            ps.setLong(1, current);
            ps.setString(2, Tool.DEFAULT_TIME_FORMATTER.format(LocalDateTime.now(Tool.zoneId)));
            ps.setString(3, tableName);
        });
    }

    public static void updateHbaseRecordSeqValue(int tableId, long current) {
        String sql = "UPDATE sdp_kmga_cs_hbase_log_topic_meta_table SET current_value=?, update_time=? WHERE table_id = ?";
        SQLHelper.save(sql, ps -> {
            ps.setLong(1, current);
            ps.setString(2, Tool.DEFAULT_TIME_FORMATTER.format(LocalDateTime.now(Tool.zoneId)));
            ps.setInt(3, tableId);
        });
    }

    public static long findHbaseRecordSeqValue(String tableName) {
        String sql = "SELECT current_value FROM sdp_kmga_cs_hbase_log_topic_meta_table WHERE table_name = ?";
        return SQLHelper.find(sql, ps -> {
            ps.setString(1, tableName);
            return ps.executeQuery();
        }, rs -> {
            if (rs.next()) {
                return rs.getLong(1);
            }
            return 0L;
        }).orElse(0L);
    }


    /**
     * 读取在审计日志过滤规则
     *
     * @return
     */
    public static List<AuditRule> findAllEnabledRules() {
        String sql = "SELECT ID, RESRC_ID, TYPE, NAME, DB_NAME, TBL_NAME, LEVEL, STAT, " +
                "ALERT_WAY, RECEIVER, REG_USR, REG_TIME, UPDATE_USR, UPDATE_TIME FROM sdp_busi_secu_rule WHERE stat=1";
        Map<Integer, AuditRule> rules = new HashMap<>(128);
        SQLHelper.find(sql, rs -> {
            while (rs.next()) {
                AuditRule auditRule = new AuditRule();
                auditRule.setRuleId(rs.getInt("ID"));
                auditRule.setResId(rs.getInt("RESRC_ID"));
                auditRule.setType(rs.getInt("TYPE"));
                auditRule.setName(rs.getString("NAME"));
                auditRule.setDbName(rs.getString("DB_NAME"));
                auditRule.setTblName(rs.getString("TBL_NAME"));
                auditRule.setRiskLevel(rs.getInt("LEVEL"));
                auditRule.setStat(rs.getInt("STAT"));
                auditRule.setAlertWay(rs.getInt("ALERT_WAY"));
                auditRule.setReceiver(rs.getString("RECEIVER"));
                auditRule.setRegUser(rs.getString("REG_USR"));
                auditRule.setRegTime(rs.getString("REG_TIME"));
                auditRule.setUpdateUser(rs.getString("UPDATE_USR"));
                auditRule.setUpdateTime(rs.getString("UPDATE_TIME"));
                rules.put(auditRule.getRuleId(), auditRule);
            }
            return null;
        });

        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("HH:MM");
        sql = "SELECT c.ID AS CON_ID, c.RULE_ID, c.IPSET_ID, c.IPSET_CMP, c.WEEK_DAYS, c.DAY_START, c.DAY_END, " +
                "c.OPER_TYPE FROM sdp_busi_secu_rule_cond c";
        Map<Integer, List<AuditRuleCondition>> ruleCons = new HashMap<>(128);
        SQLHelper.find(sql, rs -> {
            while (rs.next()) {
                AuditRuleCondition rc = new AuditRuleCondition();
                List<IPSet> ipSets = new ArrayList<>();
                rc.setIpSets(ipSets);
                rc.setId(rs.getString("CON_ID"));
                rc.setRuleId(rs.getInt("RULE_ID"));
                rc.setIpSetId(rs.getInt("IPSET_ID"));
                rc.setIpsetCmp(rs.getInt("IPSET_CMP"));

                String weekDays = rs.getString("WEEK_DAYS");
                Set<DayOfWeek> dayOfWeeks = new HashSet<>();
                if ("0".equals(weekDays) || weekDays == null) {
                    dayOfWeeks.add(DayOfWeek.MONDAY);
                    dayOfWeeks.add(DayOfWeek.TUESDAY);
                    dayOfWeeks.add(DayOfWeek.WEDNESDAY);
                    dayOfWeeks.add(DayOfWeek.THURSDAY);
                    dayOfWeeks.add(DayOfWeek.FRIDAY);
                    dayOfWeeks.add(DayOfWeek.SATURDAY);
                    dayOfWeeks.add(DayOfWeek.SUNDAY);
                } else {
                    String[] weekDayArr = weekDays.split(",");
                    for (String day : weekDayArr) {
                        dayOfWeeks.add(Tool.formatWeek(day));
                    }
                }
                rc.setWeekDays(dayOfWeeks);
                String start = rs.getString("DAY_START");
                if (start != null && !start.isEmpty()) {
                    rc.setDayStart(LocalTime.parse(start, formatter));
                }
                String end = rs.getString("DAY_END");
                if (start != null && !end.isEmpty()) {
                    rc.setDayEnd(LocalTime.parse(end, formatter));
                }
                String opTypeStr = rs.getString("OPER_TYPE");
                if (opTypeStr != null && !opTypeStr.isEmpty()) {
                    String[] opArr = opTypeStr.split(",");
                    Set<String> set = new HashSet<>();
                    for (String ip : opArr) {
                        set.add(ip);
                    }
                    rc.setOperationType(set);
                }

                String ipSetIds = rs.getString("IPSET_ID");
                if (ipSetIds != null && !ipSetIds.isEmpty()) {
                    String[] ipIds = ipSetIds.split(",");
                    for (String id : ipIds) {
                        String tmp = "SELECT i.ID AS IPSET_ID, i.NAME, i.IP_SET, i.IP_SEGM FROM sdp_busi_secu_rule_ipset i WHERE i.ID = ?";

                        SQLHelper.find(tmp, ps -> {
                            ps.setString(1, id);
                            return ps.executeQuery();
                        }, irs -> {
                            if (irs.next()) {
                                IPSet ipSet = new IPSet();
                                ipSet.setId(irs.getString("IPSET_ID"));
                                ipSet.setName(irs.getString("NAME"));
                                String setStr = irs.getString("IP_SET");
                                if (setStr != null) {
                                    String[] ipArr = setStr.split(",");
                                    Set<String> set = new HashSet<>();
                                    for (String ip : ipArr) {
                                        set.add(ip);
                                    }
                                    ipSet.setIpSet(set);
                                }
                                String ipSegm = irs.getString("IP_SEGM");
                                List<IPSet.IpSeg> ipSegs = new ArrayList<>();
                                if (ipSegm != null && !ipSegm.isEmpty()) {
                                    String[] segs = ipSegm.split(",");
                                    for (String seg : segs) {
                                        String[] arr = seg.split("[-~_|]");
                                        if (arr.length == 2) {
                                            ipSegs.add(new IPSet.IpSeg(arr[0], arr[1]));
                                        }
                                    }
                                }
                                ipSet.setIpSegs(ipSegs);
                                ipSets.add(ipSet);
                            }
                            return null;
                        });
                    }
                }

                List<AuditRuleCondition> auditRuleConditions = ruleCons.computeIfAbsent(rc.getRuleId(), k -> new ArrayList<>());
                auditRuleConditions.add(rc);
            }
            return null;
        });

        List<AuditRule> res = new ArrayList<>();
        for (Map.Entry<Integer, AuditRule> ruleEntry : rules.entrySet()) {
            Integer rid = ruleEntry.getKey();
            AuditRule auditRule = ruleEntry.getValue();
            auditRule.setAuditRuleConditions(ruleCons.get(rid));
            res.add(auditRule);
        }

        return res;
    }

    /**
     * 查找资源
     * @param ip 服务器ip
     * @param port 端口号
     * @return
     */
    public static ResourceAuditObj findResourceIDByKey(String ip, int port) {
        String sql = "SELECT a.ID, a.RESRC_ID, a.NAME, a.SERVER_IP, a.PORT, a.TYPE, a.SRC, a.ALIAS, a.REMARK, a.GET_TIME, b.RESRC_TYPE_CODE " +
                "FROM sdp_busi_resrc_audit_obj a LEFT JOIN sdp_busi_resrc b ON a.RESRC_ID = b.ID WHERE a.SERVER_IP=? AND a.PORT=?";
        return SQLHelper.find(sql, ps -> {
            ps.setString(1, ip);
            ps.setInt(2, port);
            return ps.executeQuery();
        }, rs -> {
            if (rs.next()) {
                ResourceAuditObj obj = new ResourceAuditObj();
                obj.setId(rs.getInt("ID"));
                obj.setResrcId(rs.getInt("RESRC_ID"));
                obj.setName(rs.getString("NAME"));
                obj.setServerIp(rs.getString("SERVER_IP"));
                obj.setPort(rs.getInt("PORT"));
                obj.setType(rs.getString("TYPE"));
                obj.setSrc(rs.getInt("SRC"));
                obj.setAlias(rs.getString("ALIAS"));
                obj.setRemark(rs.getString("REMARK"));
                obj.setGetTime(rs.getString("GET_TIME"));
                obj.setResTypeCode(rs.getString("RESRC_TYPE_CODE"));
                return obj;
            }
            return null;
        }).orElse(null);
    }

    /**
     * 获取全部客户端集群信息
     * @return
     */
    public static Map<String, int[]> findAllClientCluster() {
        String sql = "SELECT ip, cluster_id, CLUSTER.type FROM sdp_sys_client_ip AS IP INNER JOIN sdp_sys_client_cluster AS CLUSTER ON IP.cluster_id = CLUSTER.id";
        Map<String, int[]> res = new HashMap<>(128);
        SQLHelper.find(sql, rs -> {
           while (rs.next()) {
               int[] cluster = new int[2];
               cluster[0] = rs.getInt("cluster_id");
               cluster[1] = rs.getInt("type");
               String ip = rs.getString("ip");
               res.put(ip, cluster);
           }
           return null;
        });
        return res;
    }

    /**
     * 查找资源
     * @param src 来源
     * @return
     */
    public static List<ResourceAuditObj> findResourceBySrc(int src) {
        String sql = "SELECT a.ID, a.RESRC_ID, a.NAME, a.SERVER_IP, a.PORT, a.TYPE, a.SRC, a.ALIAS, a.REMARK, a.GET_TIME, b.RESRC_TYPE_CODE " +
                "FROM sdp_busi_resrc_audit_obj a LEFT JOIN sdp_busi_resrc b ON a.RESRC_ID = b.ID WHERE a.SRC=?";

        List<ResourceAuditObj> res = new ArrayList<>(1024);

        SQLHelper.find(sql, ps -> {
            ps.setInt(1, src);
            return ps.executeQuery();
        }, rs -> {
            while (rs.next()) {
                ResourceAuditObj obj = new ResourceAuditObj();
                obj.setId(rs.getInt("ID"));
                obj.setResrcId(rs.getInt("RESRC_ID"));
                obj.setName(rs.getString("NAME"));
                obj.setServerIp(rs.getString("SERVER_IP"));
                obj.setPort(rs.getInt("PORT"));
                obj.setType(rs.getString("TYPE"));
                obj.setSrc(rs.getInt("SRC"));
                obj.setAlias(rs.getString("ALIAS"));
                obj.setRemark(rs.getString("REMARK"));
                obj.setGetTime(rs.getString("GET_TIME"));
                obj.setResTypeCode(rs.getString("RESRC_TYPE_CODE"));
                res.add(obj);
            }
            return null;
        });
        return res;
    }


    /**
     * 查找全部资源
     * @return
     */
    public static List<ResourceAuditObj> findAllResource() {
        String sql = "SELECT a.ID, a.RESRC_ID, a.NAME, a.SERVER_IP, a.PORT, a.TYPE, a.SRC, a.ALIAS, a.REMARK, a.GET_TIME, b.RESRC_TYPE_CODE " +
                "FROM sdp_busi_resrc_audit_obj a LEFT JOIN sdp_busi_resrc b ON a.RESRC_ID = b.ID";
        List<ResourceAuditObj> res = new ArrayList<>(1024);
        SQLHelper.find(sql, rs -> {
            while (rs.next()) {
                ResourceAuditObj obj = new ResourceAuditObj();
                obj.setId(rs.getInt("ID"));
                obj.setResrcId(rs.getInt("RESRC_ID"));
                obj.setName(rs.getString("NAME"));
                obj.setServerIp(rs.getString("SERVER_IP"));
                obj.setPort(rs.getInt("PORT"));
                obj.setType(rs.getString("TYPE"));
                obj.setSrc(rs.getInt("SRC"));
                obj.setAlias(rs.getString("ALIAS"));
                obj.setRemark(rs.getString("REMARK"));
                obj.setGetTime(rs.getString("GET_TIME"));
                obj.setResTypeCode(rs.getString("RESRC_TYPE_CODE"));
                res.add(obj);
            }
            return null;
        });

        return res;
    }

    /**
     * 获取操作码表
     * @return
     */
    public static Map<String, OpCode> findAllOpCode() {
        String sql = "SELECT oper, oper_type_code, direction, resrc_type_code FROM sdp_sys_oper_classify";
        Map<String, OpCode> map = new HashMap<>(16);
        SQLHelper.find(sql, rs -> {
           while (rs.next()) {
               String oper = rs.getString("oper");
               String operTypeCode = rs.getString("oper_type_code");
               int direction = rs.getInt("direction");
               int resrcTypeCode = rs.getInt("resrc_type_code");
               OpCode code = new OpCode();
               code.setDirection(direction);
               code.setOper(oper);
               code.setOperTypeCode(operTypeCode);
               code.setResrcTypeCode(resrcTypeCode);
               map.put(oper.toLowerCase(), code);
           }
           return null;
        });
        return map;
    }

    public static Map<Integer, Integer> findAllRiskMap() {
        String sql = "SELECT ORIG_RISK_LEV, LEVEL FROM sdp_kmga_cs_risk_level_map";
        Map<Integer, Integer> map = new HashMap<>();
        SQLHelper.find(sql, rs -> {
            while (rs.next()) {
                map.put(rs.getInt("ORIG_RISK_LEV"), rs.getInt("LEVEL"));
            }
            return null;
        });
        return map;
    }

    /**
     *
     * @param briefInfos
     * @param ct create time
     */
    public static void saveAuditTrafficInfoBatch(List<AuditBriefInfo> briefInfos, LocalDateTime ct) {
        String sql = "INSERT INTO sdp_busi_rt_traffic_rcd_resrc(resrc_id, resrc_type_code, client_ip, cluster_id, out_num, in_num" +
                ", access_num, src, create_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";
        SQLHelper.saveBatch(sql, pstmt -> {
            String createTime = Tool.DEFAULT_TIME_FORMATTER.format(ct);
            for(AuditBriefInfo info : briefInfos) {
                if (info.getInNum() > 0 || info.getInAccessNum() > 0) {
                    pstmt.setInt(1, info.getResrcId());
                    pstmt.setString(2, info.getResrcTypeCode());
                    pstmt.setString(3, info.getClientIp());
                    pstmt.setString(4, info.getClusterId());
                    pstmt.setInt(5, 0);
                    pstmt.setInt(6, info.getInNum());
                    pstmt.setInt(7, info.getInAccessNum());
                    pstmt.setInt(8, info.getSrc());
                    pstmt.setString(9, createTime);
                    pstmt.addBatch();
                }
                if (info.getOutNum() > 0 || info.getOutAccessNum() > 0) {
                    pstmt.setInt(1, info.getResrcId());
                    pstmt.setString(2, info.getResrcTypeCode());
                    pstmt.setString(3, info.getClientIp());
                    pstmt.setString(4, info.getClusterId());
                    pstmt.setInt(5, info.getOutNum());
                    pstmt.setInt(6, 0);
                    pstmt.setInt(7, info.getOutAccessNum());
                    pstmt.setInt(8, info.getSrc());
                    pstmt.setString(9, createTime);
                    pstmt.addBatch();
                }
            }
        });
    }

    public static void saveAuditTrafficAppInfoBatch(List<AuditBriefInfo> briefInfos) {
        String sql = "INSERT INTO sdp_busi_rt_traffic_rcd_app(resrc_id, resrc_type_code, client_ip, cluster_id, out_num, in_num" +
                ", oper_type_code, src, time, create_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
        SQLHelper.saveBatch(sql, pstmt -> {
            for(AuditBriefInfo info : briefInfos) {
                pstmt.setInt(1, info.getResrcId());
                pstmt.setString(2, info.getResrcTypeCode());
                pstmt.setString(3, info.getClientIp());
                pstmt.setString(4, info.getClusterId());
                pstmt.setInt(5, info.getOutNum());
                pstmt.setInt(6, info.getInNum());
                pstmt.setString(7, info.getOperTypeCode());
                pstmt.setInt(8, info.getSrc());
                pstmt.setString(9, info.getTime());
                pstmt.setString(10, info.getCreateTime());
                pstmt.addBatch();
            }
        });
    }

    public static Map<String, Integer> findAllAuditHbaseType(String tableName) {
        String sql = "SELECT content, seq_id FROM " + tableName;
        Map<String, Integer> res = new HashMap<>(1024);
        SQLHelper.find(sql, result -> {
            while (result.next()) {
                res.put(result.getString(1), result.getInt(2));
            }
            return null;
        });
        return res;
    }

    public static Integer findAuditHbaseTypeByContent(String tableName, String content) {
        String sql = "SELECT content, seq_id FROM " + tableName + " WHERE content=?";
        return SQLHelper.find(sql, stmt -> {
            stmt.setString(1, content);
            return stmt.executeQuery();
        }, result -> {
            int res = 0;
            while (result.next()) {
                 res = result.getInt("seq_id");
            }
            if (res != 0) {
                return res;
            } else {
                return null;
            }
        }).orElse(null);
    }

    public static void updateAuditResSeq(String tableName, Set<Integer> seqs) {
        String sql = "UPDATE " + tableName + " SET update_time = ? WHERE seq_id = ?";
        SQLHelper.saveBatch(sql, stmt -> {
            String time = Tool.DEFAULT_TIME_FORMATTER.format(LocalDateTime.now(Tool.zoneId));
            for (Integer seq : seqs) {
                stmt.setString(1, time);
                stmt.setInt(2, seq);
                stmt.addBatch();
            }
        });
    }

    public static Integer findAuditHbaseMaxId(int typeId, String dbType) {
        String sql = "SELECT seq_id FROM sdp_busi_audit_maxid WHERE type_id=? AND db_type=?";
        return SQLHelper.find(sql, ps-> {
            ps.setInt(1, typeId);
            ps.setString(2, dbType);
            return ps.executeQuery();
        }, rs -> {
            while (rs.next()) {
                return rs.getInt(1);
            }
            return 0;
        }).orElse(0);
    }

    public static void saveErrorLog(String logType, String log) {
        String sql = "INSERT INTO sdp_kmga_cs_error_log(logtype, content) VALUES (?, ?)";
        SQLHelper.save(sql, ps -> {
           ps.setString(1, logType);
           ps.setString(2, log);
        });
    }

    /**
     * 以事务方式生成id
     * @param tableName
     * @param content
     * @param typeId
     * @param dbType
     * @return 成功返回id，失败返回-1
     */
    public static int generateAuditId(String tableName, String content, int typeId, String dbType) {
        Connection conn = null;
        try{
            conn = SQLHelper.getConnection();
            conn.setAutoCommit(false);
            String sql = "SELECT seq_id FROM sdp_busi_audit_maxid WHERE type_id=? AND db_type=?";
            String sql2 = "UPDATE sdp_busi_audit_maxid SET seq_id = ? WHERE seq_id = ? AND type_id=? AND db_type=?";
            String sql3 = "INSERT INTO "+ tableName + "(content, seq_id, create_time, update_time) VALUES(?, ?, ?, ?)";
            PreparedStatement pst = conn.prepareStatement(sql);
            pst.setInt(1, typeId);
            pst.setString(2, dbType);
            ResultSet rs = pst.executeQuery();
            while (rs.next()) {
                int seq = rs.getInt(1);

                PreparedStatement pst2 = conn.prepareStatement(sql2);
                pst2.setInt(1, seq+1);
                pst2.setInt(2, seq);
                pst2.setInt(3, typeId);
                pst2.setString(4, dbType);

                boolean bool = pst2.executeUpdate() == 1;

                if (bool) {
                    String time = Tool.DEFAULT_TIME_FORMATTER.format(LocalDateTime.now(Tool.zoneId));
                    PreparedStatement pst3 = conn.prepareStatement(sql3);
                    pst3.setString(1, content);
                    pst3.setInt(2, seq+1);
                    pst3.setString(3, time);
                    pst3.setString(4, time);
                    bool = pst3.executeUpdate() == 1;
                    if (bool) {
                        conn.commit();
                        pst.close();
                        pst2.close();
                        pst3.close();
                        return seq+1;
                    }
                    pst3.close();
                } else {
                    conn.rollback();
                }
                pst2.close();
            }
            pst.close();
        } catch (SQLException e) {
            if (conn != null) {
                try {
                    conn.rollback();
                } catch (SQLException e1) {
                    e1.printStackTrace();
                }
            }
            e.printStackTrace();
        } finally {
            if (conn != null) {
                try {
                    conn.setAutoCommit(true);
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
        return -1;
    }

    public static final int AUDIT_TYPE_ID_CLIENT_IP = 1;
    public static final int AUDIT_TYPE_ID_DB_ACCOUNT = 2;
    public static final int AUDIT_TYPE_ID_DB_NAME = 3;
    public static final int AUDIT_TYPE_ID_TABLE_NAME = 4;

    public static final String MYSQL_AUDIT_TYPE_TABLE_CLIENT_IP = "sdp_busi_audit_mysql_type_1";
    public static final String MYSQL_AUDIT_TYPE_TABLE_DB_ACCOUNT = "sdp_busi_audit_mysql_type_2";
    public static final String MYSQL_AUDIT_TYPE_TABLE_DB_NAME = "sdp_busi_audit_mysql_type_3";
    public static final String MYSQL_AUDIT_TYPE_TABLE_TABLE_NAME = "sdp_busi_audit_mysql_type_4";

    public static final String HBASE_AUDIT_TYPE_TABLE_CLIENT_IP = "sdp_busi_audit_hbase_type_1";
    public static final String HBASE_AUDIT_TYPE_TABLE_DB_ACCOUNT = "sdp_busi_audit_hbase_type_2";
    public static final String HBASE_AUDIT_TYPE_TABLE_DB_NAME = "sdp_busi_audit_hbase_type_3";
    public static final String HBASE_AUDIT_TYPE_TABLE_TABLE_NAME = "sdp_busi_audit_hbase_type_4";

    public static final String SOLR_AUDIT_TYPE_TABLE_CLIENT_IP = "sdp_busi_audit_solr_type_1";
    public static final String SOLR_AUDIT_TYPE_TABLE_DB_ACCOUNT = "sdp_busi_audit_solr_type_2";
    public static final String SOLR_AUDIT_TYPE_TABLE_DB_NAME = "sdp_busi_audit_solr_type_3";
    public static final String SOLR_AUDIT_TYPE_TABLE_TABLE_NAME = "sdp_busi_audit_solr_type_4";

    public static final String MONGODB_AUDIT_TYPE_TABLE_CLIENT_IP = "sdp_busi_audit_mongodb_type_1";
    public static final String MONGODB_AUDIT_TYPE_TABLE_DB_ACCOUNT = "sdp_busi_audit_mongodb_type_2";
    public static final String MONGODB_AUDIT_TYPE_TABLE_DB_NAME = "sdp_busi_audit_mongodb_type_3";
    public static final String MONGODB_AUDIT_TYPE_TABLE_TABLE_NAME = "sdp_busi_audit_mongodb_type_4";

    public static final String FTP_AUDIT_TYPE_TABLE_CLIENT_IP = "sdp_busi_audit_ftp_type_1";
    public static final String FTP_AUDIT_TYPE_TABLE_DB_ACCOUNT = "sdp_busi_audit_ftp_type_2";
    public static final String FTP_AUDIT_TYPE_TABLE_DB_NAME = "sdp_busi_audit_ftp_type_3";
    public static final String FTP_AUDIT_TYPE_TABLE_TABLE_NAME = "sdp_busi_audit_ftp_type_4";
}
